﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace main
{
    /// <summary>
    /// 一个基于BlockingCollection实现的多线程的处理队列
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public class ProcessQueue<T>
    {
        /// <summary>
        /// 阻塞集合
        /// </summary>
        private BlockingCollection<T> _queue;
        /// <summary>
        /// 向应该被取消的CancellationToken发送信号
        /// </summary>
        private CancellationTokenSource _cancellationTokenSource;
        /// <summary>
        /// 传播有关应取消操作的通知
        /// </summary>
        private CancellationToken _cancellationToken;

        /// <summary>
        /// 内部线程池
        /// </summary>
        private List<Thread> _threadCollection;
        /// <summary>
        /// 队列是否在处理数据
        /// </summary>
        private int _isProcessing;
        /// <summary>
        /// 有线程在处理数据
        /// </summary>
        private const int Processing = 1;
        /// <summary>
        /// 没有线程处理数据
        /// </summary>
        private const int UnProcessing = 0;
        /// <summary>
        /// 队列是否可用,volatile多线程中可被修改和每次读取值，不会因为编译器而优化
        /// </summary>
        private volatile bool _enabled = true;
        /// <summary>
        /// 内部处理线程数量
        /// </summary>
        private int _internalThreadCount;

        /// <summary>
        /// 对队列中的每个元素进行处理
        /// </summary>
        public event Action<T> ProcessItemEvent;
        /// <summary>
        /// 处理异常，需要三个参数，当前队列实例，异常，当时处理的数据
        /// </summary>
        public event Action<dynamic, Exception, T> ProcessExceptionEvent;

        /// <summary>
        /// 构造函数
        /// </summary>
        public ProcessQueue()
        {
            _queue = new BlockingCollection<T>();
            _cancellationTokenSource = new CancellationTokenSource();
            _internalThreadCount = 1;
            _cancellationToken = _cancellationTokenSource.Token;        // 关联
            _threadCollection = new List<Thread>();
        }

        /// <summary>
        /// 带参数的构造函数
        /// </summary>
        /// <param name="internalThreadCount">内部线程数量</param>
        public ProcessQueue(int internalThreadCount) : this()
        {
            this._internalThreadCount = internalThreadCount;
        }

        /// <summary>
        /// 获取内部元素的数量
        /// </summary>
        /// <returns></returns>
        public int GetInternalItemCount()
        {
            return _queue.Count;
        }

        /// <summary>
        /// 入队
        /// </summary>
        /// <param name="items"></param>
        public void Enqueue(T items)
        {
            if (items == null)
            {
                throw new ArgumentException("items");
            }
            _queue.Add(items);
            DataAdded();
        }

        /// <summary>
        /// 数据添加
        /// </summary>
        private void DataAdded()
        {
            if (_enabled)
            {
                if (!IsProcessingItem())
                {
                    ProcessRangeItem();
                    StartProcess();
                }
            }
        }

        /// <summary>
        /// 判断队列是否有线程正在处理
        /// </summary>
        /// <returns></returns>
        private bool IsProcessingItem()
        {
            return !(Interlocked.CompareExchange(ref _isProcessing, Processing, UnProcessing) == UnProcessing);
        }

        /// <summary>
        /// 启动线程处理
        /// </summary>
        private void StartProcess()
        {
            foreach(var thread in _threadCollection)
            {
                thread.Start();
            }
        }

        /// <summary>
        /// 处理范围内的项目
        /// </summary>
        private void ProcessRangeItem()
        {
            for(int i = 0; i < this._internalThreadCount; i++)
            {
                ProcessItem();
            }
        }

        /// <summary>
        /// 处理项目
        /// </summary>
        private void ProcessItem()
        {
            Thread currentThread = new Thread((state) => {
                T item = default(T);
                while (_enabled)
                {
                    try
                    {
                        try
                        {
                            item = _queue.Take(_cancellationToken);
                            ProcessItemEvent(item);
                        }
                        catch (OperationCanceledException e)
                        {
                            Console.WriteLine(e.ToString());
                        }
                    }
                    catch (Exception ex)
                    {
                        OnProcessException(ex, item);
                    }
                }
            });
            _threadCollection.Add(currentThread);
        }

        /// <summary>
        /// 清空缓存
        /// </summary>
        public void Flush()
        {
            StopProcess();

            while(_queue.Count != 0)
            {
                T item = default(T);
                if (_queue.TryTake(out item))
                {
                    try
                    {
                        ProcessItemEvent(item);
                    }
                    catch (Exception ex)
                    {
                        OnProcessException(ex, item);
                    }
                }
            }
        }

        /// <summary>
        /// 处理异常
        /// </summary>
        /// <param name="ex"></param>
        /// <param name="item"></param>
        private void OnProcessException(Exception ex, T item)
        {
            var tempException = ProcessExceptionEvent;
            Interlocked.CompareExchange(ref ProcessExceptionEvent, null, null);

            if (tempException != null)
            {
                ProcessExceptionEvent(this, ex, item);
            }
        }

        /// <summary>
        /// 停止线程处理
        /// </summary>
        private void StopProcess()
        {
            this._enabled = false;
            foreach (var thread in _threadCollection)
            {
                if (thread.IsAlive)
                {
                    thread.Join();
                }
            }
            _threadCollection.Clear();
        }
    }
}
